/**
 * 
 */
package dm.ass2.user;

import java.io.IOException;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import dm.ass2.ClusteringDriver;
import dm.ass2.io.CodebookUsersWritable;
import dm.ass2.user.io.CBElement;
import dm.ass2.user.io.UserMapOutputValueWritable;

/**
 * @author yaron
 */
public class UserReducer extends Reducer<IntWritable, UserMapOutputValueWritable, Text, CodebookUsersWritable> {
	List<CodebookUsersWritable> cbUsers;
	int numOfUserClusters = 2;
	int numOfItemClusters = 2;

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		super.cleanup(context);
		writeCbUsers(ClusteringDriver.OUTPUT_DIR + ClusteringDriver.CB_USERS_FILENAME, false, context);
	}

	/**
	 * 
	 * @param filename
	 * @param text a text file (false means binary)
	 * @param context
	 * @throws IOException
	 */
	private void writeCbUsers(String filename, boolean text, Context context) throws IOException {
		Path cbUsersPath = new Path(filename);
		FileSystem fs;

		fs = FileSystem.get(context.getConfiguration());

		if (fs.exists(cbUsersPath)) {
			fs.delete(cbUsersPath, true);
		}

		FSDataOutputStream out = fs.create(cbUsersPath);
		for (CodebookUsersWritable cbUser : cbUsers) {
			if (!text) {
				cbUser.write(out);				
			} else {
				out.writeUTF(cbUser.toString() + "\n");
			}
			
		}
		out.flush();
		out.close();
	}

	@Override
	protected void reduce(IntWritable userClusterId, Iterable<UserMapOutputValueWritable> values, Context context) throws IOException, InterruptedException {
		Set<Integer> userIds = new HashSet<Integer>();
		CBElement[] ratings = new CBElement[numOfItemClusters];
		
		for (UserMapOutputValueWritable userMapOutput : values) {
			for (int currUserId : userMapOutput.getUsersList()) {
				userIds.add(currUserId);
			}

			CBElement[] userCbElements = userMapOutput.getCbElements();
			for (int i = 0; i < userCbElements.length; i++) {
				CBElement currElement;
				if (ratings[i] == null) {
					currElement = new CBElement(0, 0f);
					ratings[i] = currElement;
				} else {
					currElement = ratings[i];
				}
				currElement.increment(userCbElements[i].getRatingsCounter());
				currElement.addToSum(userCbElements[i].getRatingsSum());
			}
		}
		
		float[] clusterItemsRatings = new float[ratings.length];
		
		for (int i = 0; i < ratings.length; i++) {
			clusterItemsRatings[i] = ratings[i].avg();
		}
		
		int[] userIdsArray = ArrayUtils.toPrimitive(userIds.toArray(new Integer[userIds.size()]));
		
		CodebookUsersWritable cbUser = new CodebookUsersWritable(userClusterId.get(), userIdsArray, clusterItemsRatings);
		cbUsers.add(cbUser);
	}

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		super.setup(context);
		
		this.numOfItemClusters = Integer.parseInt(context.getConfiguration().get(UserDriver.NUM_OF_ITEM_CLUSTERS));
		this.numOfUserClusters = Integer.parseInt(context.getConfiguration().get(UserDriver.NUM_OF_USER_CLUSTERS));
		
		cbUsers = new LinkedList<CodebookUsersWritable>();
	}
	
	
}
